package mr

import (
	"encoding/json"
	"fmt"
	"io/ioutil"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"
)
import "log"
import "net/rpc"
import "hash/fnv"

const (
	TEMP_FILE_PRE_FIX = "intermediate"
)

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.

	for task := fetchTask(); task.Command != KILL; task = fetchTask() {
		switch task.Command {
		case MAP:
			doMap(mapf, task)
		case REDUCE:
			doReduce(reducef, task)
		case WAIT:
			time.Sleep(500 * time.Millisecond)
		}
	}
	// uncomment to send the Example RPC to the master.
	//CallExample()
}

// for sorting by key.
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

func doReduce(reducef func(string, []string) string, task *FetchTaskReply) {
	//fmt.Printf("reduce id:%v\n", task.Id)

	files, err := ioutil.ReadDir(".")
	if err != nil {
		log.Fatal(err)
	}
	intermediate := []KeyValue{}
	for _, fileInfo := range files {
		filename := fileInfo.Name()
		if couldReduce(filename, task.Id) {
			file, error := os.Open(filename)
			if error != nil {
				log.Fatal(error)
			}
			dec := json.NewDecoder(file)
			for {
				var kv KeyValue
				if err := dec.Decode(&kv); err != nil {
					break
				}
				//fmt.Printf("%v->%v,%+q\n",task.Id,filename,intermediate)
				intermediate = append(intermediate, kv)
			}
		}

	}

	sort.Sort(ByKey(intermediate))

	oname := fmt.Sprintf("rt-%v-%v", os.Getpid(), task.Id)
	ofile, _ := os.Create(oname)

	//
	// call Reduce on each distinct key in intermediate[],
	// and print the result to mr-out-0.
	//
	i := 0
	for i < len(intermediate) {
		j := i + 1
		for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, intermediate[k].Value)
		}
		output := reducef(intermediate[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

		i = j
	}

	//ofile.Seek(0, 0)
	////content, _ := ioutil.ReadAll(ofile)
	////fmt.Printf("filecontent -> \n%v\n", string(content))
	ofile.Close()

	os.Rename(oname, fmt.Sprintf("mr-out-%v", task.Id))

	submit(task.Command, task.Id)
}

func couldReduce(filename string, id int) bool {
	if strings.HasPrefix(filename, TEMP_FILE_PRE_FIX) {
		strs := strings.Split(filename, "-")
		number, err := strconv.Atoi(strs[2])
		if err != nil {
			log.Fatal(err)
		}

		if number == id {
			return true
		}
	}
	return false
}

func doMap(mapf func(string, string) []KeyValue, task *FetchTaskReply) {
	//fmt.Printf("map id:%v nReduce:%v file:%v\n", task.Id, task.NReduce, task.File)

	filename := task.File
	file, err := os.Open(filename)
	if err != nil {
		log.Fatalf("cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		log.Fatalf("cannot read %v", filename)
	}
	file.Close()
	kva := mapf(filename, string(content))

	tempFiles := []*os.File{}
	for i := 0; i < task.NReduce; i++ {
		tempFileName := fmt.Sprintf("%v-%v-%v-%v", TEMP_FILE_PRE_FIX, os.Getpid(), task.Id, i)
		os.Remove(tempFileName)
		tempfile, err := ioutil.TempFile(".", tempFileName)
		if err != nil {
			log.Fatalf("cannot open %v", tempFileName)
		}
		tempFiles = append(tempFiles, tempfile)
	}
	for _, kv := range kva {
		h := ihash(kv.Key) % task.NReduce
		enc := json.NewEncoder(tempFiles[h])
		enc.Encode(kv)
	}
	for i, file := range tempFiles {
		file.Close()
		os.Rename(file.Name(), fmt.Sprintf("intermediate-%v-%v", task.Id, i))
	}

	submit(task.Command, task.Id)
}

//
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

func submit(command, id int) *SubmitReply {
	args := &SubmitArgs{
		Command: command,
		Id:      id,
	}
	reply := &SubmitReply{}

	call("Master.Submit", args, reply)

	return reply
}

func fetchTask() *FetchTaskReply {
	args := &FetchTaskArgs{}
	reply := &FetchTaskReply{}

	call("Master.FetchTask", args, reply)

	return reply
}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
